from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text, inspect 
from common.log import LOG, LogLevel
from common.dbconfig import config
import time
import random
import aiofiles
import os
import asyncio


# 异步驱动
def get_asyncdb_url(database: str) -> str:
    return f"mysql+asyncmy://{config.user}:{config.password}@{config.host}:{config.port}/{database}?charset={config.charset}"


# 同步驱动
def get_syncdb_url(database: str) -> str:
    return f"mysql+pymysql://{config.user}:{config.password}@{config.host}:{config.port}/{database}?charset={config.charset}"


# 获得唯一字符串
def generate_unique_str():
    ns = time.time_ns()  # 纳秒时间戳
    rand = random.randint(0, 9999)
    return f"{ns}{rand:04d}"


class DBUtil:
    """
        该模块不控制session的开启和关闭，使用时要时刻保持session的开启状态
    """
    def __init__(self, session: AsyncSession):
        self.session = session
        

    async def drop_all_tables(self):
        try:
            await self.session.execute(text("SET FOREIGN_KEY_CHECKS = 0")) # 禁用外键检查
            # 1. 获取所有表名
            result = await self.session.execute(text(
                "SELECT table_name FROM information_schema.tables " \
                "WHERE table_schema = DATABASE()")
            )
            tables = [row[0] for row in result]
            # 2. 清空每个表
            for table in tables:
                await self.session.execute(text(f"DROP TABLE `{table}`"))
            # 3. 恢复外键检查
            await self.session.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
            LOG(LogLevel.INFO) << "所有表已清空！" << "\n"
        except Exception as e:
            LOG(LogLevel.ERROR) << f"删除所有表失败：{str(e)}" << "\n"
            raise
    

    async def drop_all_tables_startwith(self, head: str):
        f"""删除当前数据库中所有以 '{head}_' 开头的表"""
        try:
            # 正确的方式：在同步上下文中执行检测
            table_names = await self.session.run_sync(
                lambda sync_session: inspect(sync_session.connection()).get_table_names()
            )

            ma_tables = [name for name in table_names if name.startswith(f'{head}_')]

            for table in ma_tables:
                await self.session.execute(text(f"DROP TABLE IF EXISTS {table}"))
            LOG(LogLevel.INFO) << f"所有以 '{head}_' 开头的表已被删除！" << "\n"
        except Exception as e:
            LOG(LogLevel.ERROR) << f"删除所有以 '{head}_' 开头的表失败: {str(e)}" << "\n"
            raise


    async def get_main_contracts(self):
        """获取本地存储的所有主力合约的代码"""
        try:
            result = await self.session.execute(text("SELECT * FROM klinedb.main_contracts"))
            return result
        except Exception as e:
            LOG(LogLevel.ERROR) << f"获取主力合约代码失败: {str(e)}" << "\n"
            raise


class FileUtil:
    def __init__(self, filename: str):
        self.filename = filename
    

    async def create_file(self):
        try:
            async with aiofiles.open(self.filename, mode='w'):
                pass  # 不需要写入内容
        except Exception as e:
            LOG(LogLevel.ERROR) << f"创建文件失败: {str(e)}" << "\n"
            raise


    async def write_file(self, content: str):
        try:
            async with aiofiles.open(self.filename, mode='a') as f:
                await f.write(content)  # 异步写入内容
        except Exception as e:
            LOG(LogLevel.ERROR) << f"写入文件失败: {str(e)}" << "\n"
            raise


    async def read_file(self) -> str:
        try:
            async with aiofiles.open(self.filename, mode='r') as f:
                return await f.read()  # 异步读取内容
        except Exception as e:
            LOG(LogLevel.ERROR) << f"读取文件失败: {str(e)}" << "\n"
            raise
    

    async def delete_file(self):
        try:
            # 删除文件只有同步版本
            await asyncio.to_thread(os.remove, self.filename)
        except FileNotFoundError:
            LOG(LogLevel.ERROR) << f"文件{self.filename}不存在！" << "\n" 
            raise
        except Exception as e:
            LOG(LogLevel.ERROR) << f"删除失败: {e}" << "\n"
            raise